package cn.ac.iie.di.ban.ycl.cleaner.handler;

import cn.ac.iie.di.ban.ycl.cleaner.CleanerMain;
import cn.ac.iie.di.ban.ycl.cleaner.common.Constants;
import cn.ac.iie.di.datadock.rdata.exchange.client.v1.REMessage;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.TimeUnit;
import org.apache.commons.configuration2.Configuration;
import org.apache.commons.lang.exception.ExceptionUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * @author zmc <zhangmingcheng@iie.ac.cn>
 * @date 2016-9-8 13:53:12
 * @version V2.0
 * @desc 数据去重
 */
public class CleanerHandler implements Runnable {

	final Logger LOG = LoggerFactory.getLogger(CleanerHandler.class);
	BlockingQueue<REMessage> queue;
	FilterInterface deduplicationFilter;
	final boolean isDeduplicaitonecessary;

	public CleanerHandler(Configuration configuration, BlockingQueue queue) {
		this.queue = queue;
		deduplicationFilter = new DeduplicationFilter(configuration);
		isDeduplicaitonecessary = configuration.getBoolean(Constants.DEDUPLICATION_NECESSARY, true);
	}

	@Override
	public void run() {
		while (true) {
			try {
				REMessage message = queue.poll(3, TimeUnit.SECONDS);
				if (message == null) {
					LOG.warn("there is not message from the receiver queue!");
					continue;
				}
				LOG.info("receive messages num: {}", message.data.size());
				handle(message);
			} catch (Exception ex) {
				LOG.error("clean data has error! {}", ExceptionUtils.getFullStackTrace(ex));
			}
		}
	}

	private void handle(REMessage message) throws Exception {
		//数据去重
		if (isDeduplicaitonecessary) {
			message = deduplicationFilter.filter(message);
		}
		CleanerMain.deduplicationQueue.put(message);
	}
}
